/*
 * Copyright 2011-2012 by The Regents of the University of California Licensed
 * under the Apache License, Version 2.0 (the "License"); you may not use this
 * file except in compliance with the License. you may obtain a copy of the
 * License from
 * 
 * http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */
package edu.ucsd.db.bassarisk.query.expression.visitor;

import java.util.ArrayList;
import java.util.BitSet;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;

import edu.ucsd.db.bassarisk.exception.VisitorException;
import edu.ucsd.db.bassarisk.query.Constants;
import edu.ucsd.db.bassarisk.query.binding.BindingSet;
import edu.ucsd.db.bassarisk.query.binding.Bindings.Binding;
import edu.ucsd.db.bassarisk.query.binding.Bindings.BindingCombination;
import edu.ucsd.db.bassarisk.query.binding.Bindings.PositionBinding;
import edu.ucsd.db.bassarisk.query.binding.Bindings.RunBinding;
import edu.ucsd.db.bassarisk.query.expression.Expressions.Expression;
import edu.ucsd.db.bassarisk.query.expression.Expressions.PositionMaxSeparation;
import edu.ucsd.db.bassarisk.query.expression.Expressions.PositionMinSeparation;
import edu.ucsd.db.bassarisk.query.expression.Expressions.PositionOrdered;
import edu.ucsd.db.bassarisk.query.expression.Expressions.PositionPhrase;
import edu.ucsd.db.bassarisk.query.expression.Expressions.PositionPredExpr;
import edu.ucsd.db.bassarisk.query.expression.Expressions.PositionSeparation;
import edu.ucsd.db.bassarisk.query.expression.Expressions.PositionWindow;
import edu.ucsd.db.bassarisk.query.predicate.PositionPredicates;
import edu.ucsd.db.bassarisk.query.predicate.PositionPredicates.PositionDeltaRangePredicate;
import edu.ucsd.db.bassarisk.query.predicate.PositionPredicates.PositionPredicate;
import edu.ucsd.db.bassarisk.query.predicate.PredicateGuardTable;

public class PositionPredicateExtractionTransformation extends
    ExpressionTreeTransformation<Void> {
  private abstract class AbstractPredicateGenerator {
    abstract void addCountPredicates(int binding1id, int binding2id, BitSet guard,
                                     int count);

    abstract void addPredicates(int binding1id, int binding2id, BitSet guard);

    protected PositionPredicate applyGuard(final BitSet guard,
                                           final PositionDeltaRangePredicate dpred) {
      return guard.cardinality() == 0 ? dpred : PositionPredicates
          .createBindingConditional(guard, dpred);
    }
  }

  private final BindingSet bindings;
  private final Collection<PositionPredicate> collectedPredicates;
  private final Map<String, PredicateGuardTable> guards;

  public PositionPredicateExtractionTransformation(final Collection<PositionPredicate> collectedPredicates,
                                                   final BindingSet bindings) {
    this.collectedPredicates = collectedPredicates;
    this.bindings = bindings;
    this.guards = new HashMap<String, PredicateGuardTable>();
    Set<String> requiredGuardMaps = new HashSet<String>();
    for (Binding binding : this.bindings) {
      requiredGuardMaps.add(binding.boundSequenceClass().getName());
    }
    for (String runClassName : requiredGuardMaps) {
      this.guards.put(runClassName, new PredicateGuardTable(this.bindings,
                                                            runClassName));
    }
  }

  private void addPredicates(final String runClassName,
                             final Iterable<Binding> topBindings,
                             final boolean counting,
                             final AbstractPredicateGenerator generator) {
    PredicateGuardTable table = this.guards.get(runClassName);
    int[] guardTableCols = this.findGuardTableColumns(table, topBindings);
    for (int i = 0; i < guardTableCols.length - 1; ++i) {
      for (int j = i + 1; j < guardTableCols.length; ++j) {
        int column1 = guardTableCols[i];
        int column2 = guardTableCols[j];
        if (table.areCompatible(column1, column2)) {
          int bind1id = table.bindingAtColumn(column1);
          int bind2id = table.bindingAtColumn(column2);
          if (counting) {
            Map<BitSet, Integer> guardsBetween;
            guardsBetween = table.countGuardedBindingsBetween(column1, column2);
            for (Entry<BitSet, Integer> mapEntry : guardsBetween.entrySet()) {
              generator.addCountPredicates(bind1id, bind2id, mapEntry.getKey(),
                                           mapEntry.getValue());
            }
          }
          BitSet guard = table.minimumGuard(column1, column2);
          generator.addPredicates(bind1id, bind2id, guard);
        }
      }
    }
  }

  private void expandBinding(final Binding toExpand, final List<Binding> outputSink,
                             final BindingSet bindings) {
    if (toExpand instanceof PositionBinding || toExpand instanceof RunBinding) {
      outputSink.add(toExpand);
    } else if (toExpand instanceof BindingCombination) {
      for (Binding binding : ((BindingCombination) toExpand).getCombinedBindings()) {
        this.expandBinding(bindings.getById(binding.getId()), outputSink, bindings);
      }
    }
  }

  private int[] findGuardTableColumns(final PredicateGuardTable guardTable,
                                      final Iterable<Binding> topBindings) {
    // The binding association table only covers the top level bindings for this
    // predicate. For predicate expansion, we care about all the bindings.
    // Expand the list of bindings to include all bindings without breaking the
    // order.
    List<Binding> allBindings = new ArrayList<Binding>();
    for (Binding binding : topBindings) {
      this.expandBinding(binding, allBindings, this.bindings);
    }
    // Figure out which guard table columns these bindings correspond to.
    int[] columnNums = new int[allBindings.size()];
    for (int i = 0; i < allBindings.size(); ++i) {
      columnNums[i] = guardTable.getColumnNumber(allBindings.get(i));
      assert columnNums[i] >= 0;
    }
    return columnNums;
  }

  @Override
  public Expression visitPositionMaxSeparation(final PositionMaxSeparation expr,
                                               final Void arg)
      throws VisitorException {
    AbstractPredicateGenerator generator = new AbstractPredicateGenerator() {
      @Override
      public void addCountPredicates(final int binding1id, final int binding2id,
                                     final BitSet guard, final int count) {
        int delta = ((count + 1) * expr.getDistanceConstraint()) + 1;
        PositionDeltaRangePredicate pred =
            PositionPredicates
                .createAbsoluteDeltaLess(binding1id, binding2id, delta);
        PositionPredicateExtractionTransformation.this.collectedPredicates.add(this
            .applyGuard(guard, pred));
      }

      @Override
      public void addPredicates(final int binding1id, final int binding2id,
                                final BitSet guard) {
        PositionDeltaRangePredicate pred =
            PositionPredicates.createLessThan(binding1id, binding2id);
        PositionPredicateExtractionTransformation.this.collectedPredicates.add(this
            .applyGuard(guard, pred));
      }
    };
    this.addPredicates(expr.getRunClassName(), expr.topLevelBindings(), true,
                       generator);
    return expr.getSubExpr().accept(this, null);
  }

  @Override
  public Expression visitPositionMinSeparation(final PositionMinSeparation expr,
                                               final Void arg)
      throws VisitorException {
    AbstractPredicateGenerator generator = new AbstractPredicateGenerator() {
      @Override
      public void addCountPredicates(final int binding1id, final int binding2id,
                                     final BitSet guard, final int count) {
        int delta = ((count + 1) * expr.getDistanceConstraint()) - 1;
        PositionDeltaRangePredicate pred =
            PositionPredicates.createAbsoluteDeltaGreater(binding1id, binding2id,
                                                          delta);
        PositionPredicateExtractionTransformation.this.collectedPredicates.add(this
            .applyGuard(guard, pred));
      }

      @Override
      public void addPredicates(final int binding1id, final int binding2id,
                                final BitSet guard) {
        PositionDeltaRangePredicate pred =
            PositionPredicates.createLessThan(binding1id, binding2id);
        PositionPredicateExtractionTransformation.this.collectedPredicates.add(this
            .applyGuard(guard, pred));
      }
    };
    this.addPredicates(expr.getRunClassName(), expr.topLevelBindings(), true,
                       generator);
    return expr.getSubExpr().accept(this, null);
  }

  @Override
  public Expression visitPositionOrdered(final PositionOrdered expr, final Void arg)
      throws VisitorException {
    AbstractPredicateGenerator generator = new AbstractPredicateGenerator() {
      @Override
      public void addCountPredicates(final int binding1id, final int binding2id,
                                     final BitSet guard, final int count) {
        String msg = "Internal Error, call addPredicates with counting = false.";
        throw new RuntimeException(msg);
      }

      @Override
      public void addPredicates(final int binding1id, final int binding2id,
                                final BitSet guard) {
        PositionDeltaRangePredicate pred =
            PositionPredicates.createLessThan(binding1id, binding2id);
        PositionPredicateExtractionTransformation.this.collectedPredicates.add(this
            .applyGuard(guard, pred));
      }
    };
    this.addPredicates(Constants.PositionSpaceFlag, expr.topLevelBindings(), false,
                       generator);
    return expr.getSubExpr().accept(this, null);
  }

  @Override
  public Expression visitPositionPhrase(final PositionPhrase expr, final Void arg)
      throws VisitorException {
    AbstractPredicateGenerator generator = new AbstractPredicateGenerator() {
      @Override
      public void addCountPredicates(final int binding1id, final int binding2id,
                                     final BitSet guard, final int count) {
        int delta = count + 1;
        PositionDeltaRangePredicate pred =
            PositionPredicates.createAbsoluteDeltaEquals(binding1id, binding2id,
                                                         delta);
        PositionPredicateExtractionTransformation.this.collectedPredicates.add(this
            .applyGuard(guard, pred));
      }

      @Override
      public void addPredicates(final int binding1id, final int binding2id,
                                final BitSet guard) {
        PositionDeltaRangePredicate pred =
            PositionPredicates.createLessThan(binding1id, binding2id);
        PositionPredicateExtractionTransformation.this.collectedPredicates.add(this
            .applyGuard(guard, pred));
      }
    };
    this.addPredicates(Constants.PositionSpaceFlag, expr.topLevelBindings(), true,
                       generator);
    return expr.getSubExpr().accept(this, null);
  }

  @Override
  public Expression
      visitPositionPredExpr(final PositionPredExpr expr, final Void arg)
          throws VisitorException {
    throw VisitorException.requireSubclassImplementation();
  }

  @Override
  public Expression visitPositionSeparation(final PositionSeparation expr,
                                            final Void arg) throws VisitorException {
    AbstractPredicateGenerator generator = new AbstractPredicateGenerator() {
      @Override
      public void addCountPredicates(final int binding1id, final int binding2id,
                                     final BitSet guard, final int count) {
        int delta = (count + 1) * expr.getDistanceConstraint();
        PositionDeltaRangePredicate pred =
            PositionPredicates.createAbsoluteDeltaEquals(binding1id, binding2id,
                                                         delta);
        PositionPredicateExtractionTransformation.this.collectedPredicates.add(this
            .applyGuard(guard, pred));
      }

      @Override
      public void addPredicates(final int binding1id, final int binding2id,
                                final BitSet guard) {
        PositionDeltaRangePredicate pred =
            PositionPredicates.createLessThan(binding1id, binding2id);
        PositionPredicateExtractionTransformation.this.collectedPredicates.add(this
            .applyGuard(guard, pred));
      }
    };
    this.addPredicates(expr.getRunClassName(), expr.topLevelBindings(), true,
                       generator);
    return expr.getSubExpr().accept(this, null);
  }

  @Override
  public Expression visitPositionWindow(final PositionWindow expr, final Void arg)
      throws VisitorException {
    AbstractPredicateGenerator generator = new AbstractPredicateGenerator() {
      @Override
      public void addCountPredicates(final int binding1id, final int binding2id,
                                     final BitSet guard, final int count) {
        String msg = "Internal Error, call addPredicates with counting = false.";
        throw new RuntimeException(msg);
      }

      @Override
      public void addPredicates(final int binding1id, final int binding2id,
                                final BitSet guard) {
        PositionDeltaRangePredicate pred =
            PositionPredicates.createAbsoluteDeltaLess(binding1id, binding2id,
                                                       expr.getDistanceConstraint());
        PositionPredicateExtractionTransformation.this.collectedPredicates.add(this
            .applyGuard(guard, pred));
      }
    };
    this.addPredicates(expr.getRunClassName(), expr.topLevelBindings(), false,
                       generator);
    return expr.getSubExpr().accept(this, null);
  }
}
